package com.bsoft.rpc.basic.zookeeper;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.Arrays;
import java.util.List;

public class ZkClient {
    // 重试之间等待的初始时间
    private static final int BASE_SLEEP_TIME = 1000;
    // 最大重试次数
    private static final int MAX_RETRIES = 3;

    public void start() throws Exception {
        // 重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);

        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                // 要连接的服务器列表
                .connectString("127.0.0.1:2181")
                // 重试策略
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(10000)
                .sessionTimeoutMs(30000)
                .namespace("bsoft")
                .build();
        // 启动连接zk
        zkClient.start();

        addWatcher(zkClient);

        Thread.sleep(500);
        // 1.创建一个节点（默认持久节点）
        //zkClient.create().forPath("/test/path");

        // 2.一次性创建多个节点，定义节点属性
        zkClient.create().creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .forPath("/test/path1", "java".getBytes());

        Thread.sleep(500);
        // 3.获取和更新节点数据
        //byte[] bytes = zkClient.getData().forPath("/test/path1");
        //System.out.println(new String(bytes));
        zkClient.setData().forPath("/test/path1", "python".getBytes());

        // 4.获取节点Stat数据
        //Stat stat = new Stat();
        //zkClient.getData().storingStatIn(stat).forPath("/test/path1");
        //System.out.println(stat);

        // 5.获取节点下所有子节点
        //List<String> childs = zkClient.getChildren().forPath("/test");
        //System.out.println(Arrays.asList(childs));

        Thread.sleep(500);
        // 6.删除节点
        zkClient.delete().forPath("/test/path1");


        while (true) {
            // 死循环
        }
    }

    private void addWatcher(CuratorFramework zkClient) {
        new Thread(()->{
            // 监听一个节点下子节点的创建、删除、更新操作（自动注册）
            PathChildrenCache cache = new PathChildrenCache(zkClient, "/test", true);

            cache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                    switch (event.getType()) {
                        case CHILD_ADDED:
                            System.out.println("=====> CHILD_ADDED : " + event.getData().getPath() + "  数据:" + new String(event.getData().getData()));
                            break;
                        case CHILD_REMOVED:
                            System.out.println("=====> CHILD_REMOVED : " + event.getData().getPath() + "  数据:" + new String(event.getData().getData()));
                            break;
                        case CHILD_UPDATED:
                            System.out.println("=====> CHILD_UPDATED : " + event.getData().getPath() + "  数据:" + new String(event.getData().getData()));
                            break;
                        default:
                            break;
                    }
                }
            });

            try {
                cache.start(PathChildrenCache.StartMode.NORMAL);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }

    public static void main(String[] args) throws Exception {
        new ZkClient().start();
    }
}
